package com.xzy.rabbit.demo.publisherconfirms;

import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import com.xzy.rabbit.common.util.RabbitUtil;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeoutException;

/**
 * 发布确认模式（Publisher Confirms）
 *
 * @author xzy
 * @date 2021年9月26日
 */
@Slf4j
@NoArgsConstructor(access = AccessLevel.PRIVATE)
class Main {
    private static final int MESSAGE_COUNT = 1000;
    private static final String SINGLE_ACK_QUEUE = "ack_queue_single";
    private static final String BATCH_ACK_QUEUE = "ack_queue_batch";
    private static final String ASYNC_ACK_QUEUE = "ack_queue_async";


    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        Connection connection = RabbitUtil.getConnection();
        CountDownLatch sendMsg = new CountDownLatch(3);

        // 单个确认
        SingleACK.produce(connection, SINGLE_ACK_QUEUE, MESSAGE_COUNT, sendMsg);
        // 批量确认
        BatchACK.produce(connection, BATCH_ACK_QUEUE, MESSAGE_COUNT, sendMsg);
        // 异步确认
        AsyncACK.produce(connection, ASYNC_ACK_QUEUE, MESSAGE_COUNT, sendMsg);

        // 收尾工作：清空队列
        consume(connection, SINGLE_ACK_QUEUE);
        consume(connection, BATCH_ACK_QUEUE);
        consume(connection, ASYNC_ACK_QUEUE);

        sendMsg.await();
        log.info("单个确认 总耗时：{}", SingleACK.workTime);
        log.info("批量确认 总耗时：{}", BatchACK.workTime);
        log.info("异步确认 总耗时：{}", AsyncACK.workTime);
    }

    /**
     * 消费消息
     */
    public static void consume(Connection connection, String queueName) throws IOException {
        // 1.获取信道
        Channel channel = connection.createChannel();

        // 3.接收消息（消费）+ 手动应答
        DeliverCallback deliverCallback = (consumerTag, message) -> {
        };
        CancelCallback cancelCallback = consumerTag -> {
        };
        channel.basicConsume(queueName, true, deliverCallback, cancelCallback);
    }
}
